package com.jltx.xq.coroutinesandrxpro.coroutines

import android.util.Log
import com.jltx.xq.coroutinesandrxpro.rxjava.RxJava2Learn1
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.BroadcastChannel
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.actor
import kotlinx.coroutines.channels.produce

/**
 * @author aidenquan
 *         Created by junlintianxia on 2021/08/09.
 */
object CoroutinesLearn3 {

    /**
     * 通道不断的往下游发送数据
     */
    fun customChannelInterval() {
        GlobalScope.launch(Dispatchers.Main) {

            /**
             * 正如我们之前所说，默认情况下，通道的容量为 1。
             */
            val channel = Channel<Int>()
            Log.d(RxJava2Learn1.TAG, "customChannelInterval start")
            /**
             * 生产者，进行发送数据到通道
             * 这是一个协程，它将通过一个简单的 for 循环向通道发送五个元素
             */
            val producer = GlobalScope.launch(Dispatchers.IO) {
                for (i in 1..5) {
                    if (!isActive) {
                        break
                    }
                    channel.send(i)
                    Log.d(RxJava2Learn1.TAG, "customChannelInterval send $i")
                    delay(1000L)
                }
            }

            /**
             * 消费者
             */
            val consumer = GlobalScope.launch(Dispatchers.IO) {
                //这个 for 循环将隐式调用 channel.receive()
                for (value in channel) {
                    Log.d(RxJava2Learn1.TAG, "customChannelInterval consumer $value")
                }
            }
        }
    }


    /**
     * 通道不断的往下游发送数据
     */
    fun customBufferChannelInterval() {
        GlobalScope.launch(Dispatchers.Main) {

            /**
             * 正如我们之前所说，默认情况下，通道的容量为 1。您可以创建缓冲通道，允许发送方在暂停之前发送多个元素
             */
            val channel = Channel<Int>(Channel.CONFLATED)
            Log.d(RxJava2Learn1.TAG, "customBufferChannelInterval start")
            /**
             * 生产者，进行发送数据到通道
             * 这是一个协程，它将通过一个简单的 for 循环向通道发送五个元素
             */
            val producer = GlobalScope.launch(Dispatchers.IO) {
                for (i in 1..5) {
                    if (!isActive) {
                        break
                    }
                    channel.send(i)
                    Log.d(RxJava2Learn1.TAG, "customBufferChannelInterval send $i")
                    delay(1000L)
                }
            }

            /**
             * 消费者
             */
            val consumer = GlobalScope.launch(Dispatchers.IO) {
                //这个 for 循环将隐式调用 channel.receive()
                for (value in channel) {
                    delay(2000L)
                    Log.d(RxJava2Learn1.TAG, "customBufferChannelInterval consumer $value")
                }
            }
            Log.d(RxJava2Learn1.TAG, "customBufferChannelInterval end")
        }
    }


    /**
     * produce生产者不断的往下游发送数据
     */
    fun customProduceChannelInterval() {
        GlobalScope.launch(Dispatchers.Main) {

            /**
             * 生产者并且返回通道，进行发送数据到ReceiveChannel通道
             * 这是一个协程，它只有生产内部的代码可以将元素发送到通道，因此它可以防止其他协程在该通道上调用发送
             * Produce 是创建自定义运算符的一种非常有用的方法。
             * 当生产 CoroutineBuilder 完成执行其挂起的 lambda 并且通道中的元素已被消耗时，通道将自动关闭
             */
            val publisher = GlobalScope.produce<Int>(capacity = 2) {
                for (i in 1..5) {
                    if (!isActive) {
                        break
                    }
                    send(i)
                    Log.d(RxJava2Learn1.TAG, "customProduceChannelInterval send $i")
                    delay(1000L)
                }
            }

            /**
             * 消费者
             */
            GlobalScope.launch(Dispatchers.IO) {
                //这个 for 循环将隐式调用 channel.receive()
                for (value in publisher) {
                    delay(1000L)
                    Log.d(RxJava2Learn1.TAG, "customProduceChannelInterval consumer $value")
                }
            }
            Log.d(RxJava2Learn1.TAG, "customProduceChannelInterval end")
        }
    }


    /**
     * actor不断的往下游发送数据
     */
    fun customActorChannelInterval() {
        GlobalScope.launch(Dispatchers.Main) {

            /**
             * 演员并且返回SendChannel通道
             * 这是一个协程，只有actor创建的协程才能消费通道中的元素
             * 您可以将参与者视为接收和处理元素的邮箱。由于协程内部的代码是顺序执行的，所以一次只会处理一个元素
             */
            val actor = GlobalScope.actor<Int>() {
                for (i in channel) {
                    if (!isActive) {
                        break
                    }
                    Log.d(RxJava2Learn1.TAG, "customActorChannelInterval consumer actor $i")
                    delay(1000L)
                }
            }

            /**
             * 驱动演员发送数据
             */
            GlobalScope.launch(Dispatchers.IO) {
                for (i in 1..5) {
                    if (!isActive) {
                        break
                    }
                    actor.send(i)
                    Log.d(RxJava2Learn1.TAG, "customActorChannelInterval send $i")
                    delay(1000L)
                }
            }
            Log.d(RxJava2Learn1.TAG, "customActorChannelInterval end")
        }
    }

    /**
     * 通道不断的往下游发送数据,订阅两个消费者进行接收
     * 任何时候就只能有一个消费者进行下游数据的消费，多个消费者的协程间是相互竞争
     */
    fun customMultiReceiveChannelInterval() {
        GlobalScope.launch(Dispatchers.Main) {

            /**
             * 正如我们之前所说，默认情况下，通道的容量为 1。
             */
            val channel = Channel<Int>()
            Log.d(RxJava2Learn1.TAG, "customMultiReceiveChannelInterval start")
            /**
             * 生产者，进行发送数据到通道
             * 这是一个协程，它将通过一个简单的 for 循环向通道发送五个元素
             */
            val producer = GlobalScope.launch(Dispatchers.IO) {
                for (i in 1..5) {
                    if (!isActive) {
                        break
                    }
                    channel.send(i)
                    Log.d(RxJava2Learn1.TAG, "customMultiReceiveChannelInterval send $i")
                    delay(1000L)
                }
            }

            /**
             * 第一个消费者
             */
            val consumer = GlobalScope.launch(Dispatchers.IO) {
                //这个 for 循环将隐式调用 channel.receive()
                for (value in channel) {
                    Log.d(RxJava2Learn1.TAG, "customMultiReceiveChannelInterval consumer $value")
                }
            }

            /**
             * 第二个消费者，通过两个消费者相互抢消费对应的生产者协程的值，可以知道
             * 当您从通道中消费一个对象时，其他协程将无法获取相同的对象
             */
            val consumer1 = GlobalScope.launch(Dispatchers.IO) {
                //这个 for 循环将隐式调用 channel.receive()
                for (value in channel) {
                    Log.d(RxJava2Learn1.TAG, "customMultiReceiveChannelInterval consumer1 $value")
                }
            }
            Log.d(RxJava2Learn1.TAG, "customMultiReceiveChannelInterval end")
        }
    }

    /**
     * 通道不断的往下游发送数据
     * 调用了它的 close，它会立即停止接受新元素，也就是说这时候它的 isClosedForSend 会立即返回 true，
     * 而由于 Channel 缓冲区的存在，这时候可能还有一些元素没有被处理完，所以要等所有的元素都被读取之后
     * isClosedForReceive 才会返回 true
     */
    fun customChannelClose() {
        GlobalScope.launch(Dispatchers.Main) {

            /**
             * 开个缓冲区狭小为3
             */
            val channel = Channel<Int>(3)
            Log.d(RxJava2Learn1.TAG, "customChannelClose start")
            /**
             * 生产者，进行发送数据到通道
             * 这是一个协程，它将通过一个简单的 for 循环向通道发送五个元素
             */
            val producer = GlobalScope.launch(Dispatchers.IO) {
                for (i in 1..5) {
                    if (!isActive) {
                        break
                    }
                    channel.send(i)
                    Log.d(RxJava2Learn1.TAG, "customChannelClose send $i")
                }
                //发送完5个数据之后，调用close
                channel.close()
                Log.d(
                    RxJava2Learn1.TAG,
                    "customChannelClose channel.close.isClosedForSend: ${channel.isClosedForSend} isClosedForReceive: ${channel.isClosedForReceive}"
                )
            }

            /**
             * 消费者
             */
            val consumer = GlobalScope.launch(Dispatchers.IO) {
                //这个 for 循环将隐式调用 channel.receive()
                for (value in channel) {
                    Log.d(RxJava2Learn1.TAG, "customChannelClose consumer $value")
                    delay(1000L)
                }
                Log.d(
                    RxJava2Learn1.TAG,
                    "customChannelClose after.consuming.isClosedForSend: ${channel.isClosedForSend} isClosedForReceive: ${channel.isClosedForReceive}"
                )
            }
        }
    }

    /**
     * 如果要将同一个对象发送给多个观察者，则需要一个 BroadcastChannel
     * 类似RxJava 的广播方式：Subjects订阅者模式
     */
    fun broadcastChannelInterval() {
        GlobalScope.launch(Dispatchers.Main) {
            val broadcastChannel = BroadcastChannel<Int>(5)

            val producer = GlobalScope.launch(Dispatchers.IO) {
                List(5) {
                    if (!isActive) {
                        return@List
                    }
                    broadcastChannel.send(it)
                    Log.d(
                        RxJava2Learn1.TAG,
                        "broadcastChannelInterval send $it Thread: ${Thread.currentThread()}"
                    )
                }
                broadcastChannel.close()
                Log.d(RxJava2Learn1.TAG, "broadcastChannelInterval broadcastChannel.close")
            }

            List(3) { index ->
                GlobalScope.launch(Dispatchers.IO) {
                    val receiveChannel = broadcastChannel.openSubscription()
                    val iterator = receiveChannel.iterator()
                    while (iterator.hasNext()) {
                        val value = iterator.next()
                        Log.d(
                            RxJava2Learn1.TAG,
                            "broadcastChannelInterval receive.index [$index] value: $value Thread: ${Thread.currentThread()}"
                        )
                        delay(1000L)
                    }
                    Log.d(
                        RxJava2Learn1.TAG,
                        "broadcastChannelInterval after.consuming.isClosedForSend: ${broadcastChannel.isClosedForSend} isClosedForReceive: ${broadcastChannel.openSubscription().isClosedForReceive}"
                    )
                }
            }
        }

    }
}